package HandshakeJoin;

import java.util.List;
import java.util.ArrayList;
import org.apache.s4.core.ProcessingElement;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class QueryProcessor extends ProcessingElement
{
    transient private static final Logger logger = LoggerFactory
            .getLogger(QueryProcessor.class);

    private int numFields = 7;
    private String dbChoice;
    private DBClient db;

    @Override
    protected void onCreate() {
        db = null;
        if (dbChoice != null) {
            if (dbChoice.equals("cassandra")) {
                db = new CassandraClient("join" + getId(), numFields);
            }
            if (dbChoice.equals("hbase")) {
                db = new HBaseClient("join" + getId());
            }
            else if (dbChoice.equals("logbase")) {
                db = new LogBaseClient("join" + getId(), numFields);
            }
        }
    }

    public void onEvent(QueryRequest queryReq) {
        QueryRequest.Type queryType = queryReq.getType();
        boolean rc;

        if (queryType == QueryRequest.Type.ONE_TIME) {
            rc = processOneTimeQuery(queryReq);
        }
        else if (queryType == QueryRequest.Type.CONTINUOUS) {
            rc = processContinuousQuery(queryReq);
        }
        else {
            rc = false;
            logger.error("unknown query");
        }

        if (!rc) {
            logger.error("failed to process query: " + queryReq);
        }
    }

    @Override
    protected void onRemove() {
        if (db != null) {
            db.close();
        }
    }

    private boolean processOneTimeQuery(final QueryRequest queryReq) {
        if (db == null) {
            logger.error("no database service");
            return false;
        }

        QueryRequest.Subtype subtype = queryReq.getSubtype();
        List<ArrayList<Integer>> result = null;

        if (subtype == QueryRequest.Subtype.EQUAL_BY_JOIN_KEY) {
            int value = Integer.parseInt(queryReq.getArgument("value"));

            result = db.retrieveByJoinKey(value);
        }
        else if (subtype == QueryRequest.Subtype.RANGE_BY_JOIN_KEY) {
            int valueLower = Integer.parseInt(queryReq
                    .getArgument("valueLower"));
            int valueUpper = Integer.parseInt(queryReq
                    .getArgument("valueUpper"));

            result = db.retrieveByJoinKey(valueLower, valueUpper);
        }
        else if (subtype == QueryRequest.Subtype.EQUAL_BY_TIMESTAMP) {
            long timestamp = Integer
                    .parseInt(queryReq.getArgument("timestamp"));

            result = db.retrieveByTimestamp(timestamp);
        }
        else if (subtype == QueryRequest.Subtype.RANGE_BY_TIMESTAMP) {
            long tsLower = Integer.parseInt(queryReq
                    .getArgument("timestampLower"));
            long tsUpper = Integer.parseInt(queryReq
                    .getArgument("timestampUpper"));

            result = db.retrieveByTimestamp(tsLower, tsUpper);
        }
        else {
            logger.error("unknown subtype");
        }

        if (result != null) {
            print(queryReq.getRequestId(), result); // TEST
            return true;
        }
        else {
            return false;
        }
    }

    private boolean processContinuousQuery(final QueryRequest queryReq) {
        // TODO
        return true;
    }

    public QueryProcessor setNumberOfFields(final int numFields) {
        this.numFields = numFields;

        return this;
    }

    public QueryProcessor setDBChoice(final String dbChoice) {
        this.dbChoice = dbChoice;

        return this;
    }

    /* Print records, for testing. */
    private void print(final String title,
            final List<ArrayList<Integer>> records) {
        System.out.println(">>> " + title + " <<<");
        for (ArrayList<Integer> record : records) {
            System.out.println(record);
        }
        System.out.println(">>>--------------------<<<");
        System.out.println();
    }
}